package com.tky.utils;

import com.alibaba.fastjson2.JSONObject;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * HBase 操作工具类，使用连接池管理 HBase 连接，提供表管理、数据插入、查询和删除功能。
 * 所有方法为静态方法，无需实例化，支持线程安全操作，确保资源正确释放。
 */
public class HBaseUtil {
    private static final Logger logger = LoggerFactory.getLogger(HBaseUtil.class);
    private static final String CONFIG_PATH = "config.properties";
    private static volatile Connection connection; // 单例连接，线程安全
    private static final Configuration hbaseConf;

    // 静态初始化配置
    static {
        hbaseConf = initHBaseConfiguration();
    }

    /**
     * 初始化 HBase 配置，从 config.properties 加载参数，包括连接池设置。
     *
     * @return 配置对象
     * @throws IllegalArgumentException 如果配置文件或键缺失
     */
    private static Configuration initHBaseConfiguration() {
        Configuration conf = HBaseConfiguration.create();
        String[] configKeys = {
                "hbase.zookeeper.znode.parent",
                "hbase.zookeeper.quorum",
                "hbase.zookeeper.clientPort",
                "hbase.master",
                "hbase.rootdir",
                "hbase.client.ipc.pool.size"
        };
        for (String key : configKeys) {
            String value = PropertiesReader.getPropertyByKey(key, CONFIG_PATH);
            if (value == null && !key.equals("hbase.client.ipc.pool.size")) {
                throw new IllegalArgumentException("HBase 配置键未找到: " + key);
            }
            if (value != null) {
                conf.set(key, value);
            }
        }
        // 默认连接池大小为 10
        if (conf.get("hbase.client.ipc.pool.size") == null) {
            conf.set("hbase.client.ipc.pool.size", "10");
        }
        return conf;
    }

    /**
     * 获取 HBase 连接，使用连接池，延迟初始化，线程安全。
     *
     * @return HBase 连接
     * @throws IOException 如果连接失败
     */
    private static synchronized Connection getConnection() throws IOException {
        if (connection == null || connection.isClosed()) {
            connection = ConnectionFactory.createConnection(hbaseConf);
        }
        return connection;
    }

    /**
     * 关闭 HBase 连接。
     *
     * <p><b>使用示例：</b></p>
     * <pre>
     *     try {
     *         HBaseUtil.putCell("TEST", "row1", "info", "name", "Alice");
     *     } finally {
     *         HBaseUtil.closeConnection();
     *     }
     * </pre>
     *
     * @throws IOException 如果关闭失败
     */
    public static void closeConnection() throws IOException {
        if (connection != null && !connection.isClosed()) {
            connection.close();
            connection = null;
        }
    }

    /**
     * 验证输入参数的有效性。
     *
     * @param tableName 表名
     * @param rowKey    行键（可为 null）
     * @param family    列族（可为 null）
     * @param column    列名（可为 null）
     */
    private static void validateInputs(String tableName, String rowKey, String family, String column) {
        if (tableName == null || tableName.trim().isEmpty()) {
            throw new IllegalArgumentException("表名不能为空或空字符串");
        }
        if (rowKey != null && rowKey.trim().isEmpty()) {
            throw new IllegalArgumentException("行键不能为空字符串");
        }
        if (family != null && family.trim().isEmpty()) {
            throw new IllegalArgumentException("列族不能为空字符串");
        }
        if (column != null && column.trim().isEmpty()) {
            throw new IllegalArgumentException("列名不能为空字符串");
        }
    }

    /**
     * 判断表是否存在。
     *
     * @param tableName 表名，例如 "TEST"
     * @return true 表示表存在
     * @throws IOException 如果操作失败
     *
     *                     <p><b>使用示例：</b></p>
     *                     <pre>
     *                                             boolean exists = HBaseUtil.isTableExists("TEST");
     *                                             System.out.println("Table exists: " + exists);
     *                                         </pre>
     */
    public static boolean isTableExists(String namespace, String tableName) throws IOException {
        validateInputs(tableName, null, null, null);
        try (Connection conn = getConnection(); Admin admin = conn.getAdmin()) {
            return admin.tableExists(TableName.valueOf(namespace, tableName));
        }
    }

    /**
     * 创建表。
     *
     * @param tableName      表名，例如 "TEST"
     * @param columnFamilies 列族名称数组，例如 {"info", "data"}
     * @throws IOException 如果创建失败
     *
     *                     <p><b>使用示例：</b></p>
     *                     <pre>
     *                                             HBaseUtil.createTable("TEST", "info", "data");
     *                                         </pre>
     */
    public static void createTable(String namespace, String tableName, String... columnFamilies) throws IOException {
        validateInputs(tableName, null, null, null);
        if (columnFamilies.length == 0) {
            throw new IllegalArgumentException("创建表至少需要一个列族");
        }
        if (isTableExists(namespace, tableName)) {
            System.out.println("表: " + namespace + ":" + tableName + " 已经存在");
            return;
        }
        try (Connection conn = getConnection(); Admin admin = conn.getAdmin()) {
            TableDescriptorBuilder tableBuilder = TableDescriptorBuilder.newBuilder(
                    TableName.valueOf(namespace, tableName));
            for (String cf : columnFamilies) {
                ColumnFamilyDescriptorBuilder cfBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf));
                cfBuilder.setMaxVersions(5);
                tableBuilder.setColumnFamily(cfBuilder.build());
            }
            admin.createTable(tableBuilder.build());
        }
    }

    /**
     * 修改表中列族的版本。
     *
     * @param tableName    表名，例如 "TEST"
     * @param columnFamily 列族名称，例如 "info"
     * @param version      版本数
     * @throws IOException 如果修改失败
     *
     *                     <p><b>使用示例：</b></p>
     *                     <pre>
     *                                             HBaseUtil.modifyTable("TEST", "info", 3);
     *                                         </pre>
     */
    public static void modifyTable(String namespace, String tableName, String columnFamily, int version) throws IOException {
        validateInputs(tableName, null, columnFamily, null);
        if (!isTableExists(namespace, tableName)) {
            throw new IllegalArgumentException("表不存在: " + tableName);
        }
        try (Connection conn = getConnection(); Admin admin = conn.getAdmin()) {
            TableDescriptor descriptor = admin.getDescriptor(TableName.valueOf(namespace, tableName));
            TableDescriptorBuilder tableBuilder = TableDescriptorBuilder.newBuilder(descriptor);
            ColumnFamilyDescriptor cfDescriptor = descriptor.getColumnFamily(Bytes.toBytes(columnFamily));
            if (cfDescriptor == null) {
                logger.error("列族 {} 不存在于表 {}", columnFamily, tableName);
                throw new IllegalArgumentException("列族不存在: " + columnFamily);
            }
            ColumnFamilyDescriptorBuilder cfBuilder = ColumnFamilyDescriptorBuilder.newBuilder(cfDescriptor);
            cfBuilder.setMaxVersions(version);
            tableBuilder.modifyColumnFamily(cfBuilder.build());
            admin.modifyTable(tableBuilder.build());
        }
    }

    /**
     * 删除表。
     *
     * @param tableName 表名，例如 "TEST"
     * @return true 表示删除成功
     * @throws IOException 如果删除失败
     *
     *                     <p><b>使用示例：</b></p>
     *                     <pre>
     *                                             boolean deleted = HBaseUtil.deleteTable("TEST");
     *                                             System.out.println("Table deleted: " + deleted);
     *                                         </pre>
     */
    public static boolean deleteTable(String namespace, String tableName) throws IOException {
        validateInputs(tableName, null, null, null);
        if (!isTableExists(namespace, tableName)) {
            System.out.println("待删除的表不存在: " + namespace + ":" + tableName);
            return false;
        }
        try (Connection conn = getConnection(); Admin admin = conn.getAdmin()) {
            TableName tn = TableName.valueOf(namespace, tableName);
            admin.disableTable(tn);
            admin.deleteTable(tn);
            return true;
        }
    }

    /**
     * 插入单列数据。
     *
     * @param tableName    表名，例如 "TEST"
     * @param rowKey       行键，例如 "row1"
     * @param columnFamily 列族名称，例如 "info"
     * @param columnName   列名，例如 "name"
     * @param value        值，例如 "Alice"
     * @throws IOException 如果插入失败
     *
     *                     <p><b>使用示例：</b></p>
     *                     <pre>
     *                                             HBaseUtil.putCell("TEST", "row1", "info", "name", "Alice");
     *                                         </pre>
     */
    public static void putCell(String namespace, String tableName, String rowKey, String columnFamily, String columnName, String value) throws IOException {
        validateInputs(tableName, rowKey, columnFamily, columnName);
        try (Connection conn = getConnection(); Table table = conn.getTable(TableName.valueOf(namespace, tableName))) {
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName), Bytes.toBytes(value));
            table.put(put);
        } catch (IOException e) {
            logger.error("向表 {} 的行 {} 插入数据失败: {}", tableName, rowKey, e.getMessage(), e);
            throw e;
        }
    }

    /**
     * 查询单列数据，支持多版本。
     *
     * @param tableName    表名，例如 "TEST"
     * @param rowKey       行键，例如 "row1"
     * @param columnFamily 列族名称，例如 "info"
     * @param columnName   列名，例如 "name"
     * @return 包含所有版本值的 List<JSONObject>，每个 JSONObject 包含 "value" 和 "timestamp"
     * @throws IOException 如果查询失败
     *
     *                     <p><b>使用示例：</b></p>
     *                     <pre>
     *                                             List<JSONObject> values = HBaseUtil.getCells("TEST", "row1", "info", "name");
     *                                             for (JSONObject val : values) {
     *                                                 System.out.println(val.toJSONString());
     *                                             }
     *                                         </pre>
     */
    public static List<JSONObject> getCells(String namespace, String tableName, String rowKey, String columnFamily, String columnName) throws IOException {
        validateInputs(tableName, rowKey, columnFamily, columnName);
        List<JSONObject> results = new ArrayList<>();
        try (Connection conn = getConnection(); Table table = conn.getTable(TableName.valueOf(namespace, tableName))) {
            Get get = new Get(Bytes.toBytes(rowKey));
            get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName));
            get.readAllVersions();
            Result result = table.get(get);
            for (Cell cell : result.rawCells()) {
                JSONObject cellData = new JSONObject();
                cellData.put("value", Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                cellData.put("timestamp", cell.getTimestamp());
                results.add(cellData);
            }
            return results;
        }
    }

    /**
     * 扫描表数据，返回指定范围的行。
     *
     * @param tableName 表名，例如 "TEST"
     * @param startRow  开始行键（包含），例如 "row1"
     * @param stopRow   结束行键（不包含），例如 "row10"
     * @return 包含所有行数据的 List<JSONObject>，每个 JSONObject 包含 rowKey, family, qualifier, value
     * @throws IOException 如果扫描失败
     *
     *                     <p><b>使用示例：</b></p>
     *                     <pre>
     *                                             List<JSONObject> rows = HBaseUtil.scanRows("TEST", "row1", "row10");
     *                                             for (JSONObject row : rows) {
     *                                                 System.out.println(row.toJSONString());
     *                                             }
     *                                         </pre>
     */
    public static List<JSONObject> scanRows(String namespace, String tableName, String startRow, String stopRow) throws IOException {
        validateInputs(tableName, startRow, null, null);
        List<JSONObject> results = new ArrayList<>();
        try (Connection conn = getConnection(); Table table = conn.getTable(TableName.valueOf(namespace, tableName))) {
            Scan scan = new Scan();
            scan.withStartRow(Bytes.toBytes(startRow));
            scan.withStopRow(Bytes.toBytes(stopRow));
            scan.setCaching(1000);
            try (ResultScanner scanner = table.getScanner(scan)) {
                for (Result result : scanner) {
                    for (Cell cell : result.rawCells()) {
                        JSONObject cellData = new JSONObject();
                        cellData.put("rowKey", Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()));
                        cellData.put("family", Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));
                        cellData.put("qualifier", Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
                        cellData.put("value", Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                        results.add(cellData);
                    }
                }
                return results;
            }
        }
    }

    /**
     * 查询 HBase 表，返回指定时间范围内的所有行数据。
     *
     * @param tableName   表名，例如 "LOCOMOTIVE_ODS:CMD_LKJ_ITF_TAX"
     * @param startRowkey 开始 RowKey
     * @param endRowkey   结束 RowKey
     * @param family      列族，例如 "DATA"
     * @param columns     列名数组，例如 {"JD", "WD"}
     * @return 包含所有查询结果的 List<JSONObject>
     * @throws IOException 如果查询失败
     */
    public static List<JSONObject> queryHBaseTable(String tableName, String startRowkey, String endRowkey, String family, String[] columns) throws IOException {
        List<JSONObject> results = new ArrayList<>();
        try (Connection connection = getConnection(); Table table = connection.getTable(TableName.valueOf(tableName))) {
            Scan scan = new Scan(Bytes.toBytes(startRowkey), Bytes.toBytes(endRowkey));
            scan.setCaching(1000); // 缓存行数
            scan.addFamily(Bytes.toBytes(family));
            for (String column : columns) {
                scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));
            }
            try (ResultScanner scanner = table.getScanner(scan)) {
                for (Result res : scanner) {
                    JSONObject rowData = new JSONObject();
                    for (String column : columns) {
                        rowData.put(column, "-");
                        Cell cell = res.getColumnLatestCell(Bytes.toBytes(family), Bytes.toBytes(column));
                        if (cell != null) {
                            String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                            rowData.put(column, value);
                        }
                    }
                    results.add(rowData);
                }
            }
        }
        return results;
    }


    /**
     * 带过滤器的扫描，返回符合条件的行。
     *
     * @param tableName    表名，例如 "TEST"
     * @param startRow     开始行键（包含），例如 "row1"
     * @param stopRow      结束行键（不包含），例如 "row10"
     * @param columnFamily 列族名称，例如 "info"
     * @param columnName   列名，例如 "name"
     * @param value        过滤值，例如 "Alice"
     * @return 包含所有行数据的 List<JSONObject>，每个 JSONObject 包含 rowKey, family, qualifier, value
     * @throws IOException 如果扫描失败
     *
     *                     <p><b>使用示例：</b></p>
     *                     <pre>
     *                                             List<JSONObject> rows = HBaseUtil.filterScan("TEST", "row1", "row10", "info", "name", "Alice");
     *                                             for (JSONObject row : rows) {
     *                                                 System.out.println(row.toJSONString());
     *                                             }
     *                                         </pre>
     */
    public static List<JSONObject> filterScan(String namespace, String tableName, String startRow, String stopRow, String columnFamily, String columnName, String value) throws IOException {
        validateInputs(tableName, startRow, columnFamily, columnName);
        List<JSONObject> results = new ArrayList<>();
        try (Connection conn = getConnection(); Table table = conn.getTable(TableName.valueOf(namespace, tableName))) {
            Scan scan = new Scan();
            scan.withStartRow(Bytes.toBytes(startRow));
            scan.withStopRow(Bytes.toBytes(stopRow));
            scan.setCaching(1000);
            FilterList filterList = new FilterList();
            SingleColumnValueFilter filter = new SingleColumnValueFilter(
                    Bytes.toBytes(columnFamily),
                    Bytes.toBytes(columnName),
                    CompareOperator.EQUAL,
                    Bytes.toBytes(value));
            filterList.addFilter(filter);
            scan.setFilter(filterList);
            try (ResultScanner scanner = table.getScanner(scan)) {
                for (Result result : scanner) {
                    for (Cell cell : result.rawCells()) {
                        JSONObject cellData = new JSONObject();
                        cellData.put("rowKey", Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()));
                        cellData.put("family", Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));
                        cellData.put("qualifier", Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
                        cellData.put("value", Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                        results.add(cellData);
                    }
                }
                return results;
            }
        }
    }

    /**
     * 删除列数据（所有版本）。
     *
     * @param tableName    表名，例如 "TEST"
     * @param rowKey       行键，例如 "row1"
     * @param columnFamily 列族名称，例如 "info"
     * @param columnName   列名，例如 "name"
     * @throws IOException 如果删除失败
     *
     *                     <p><b>使用示例：</b></p>
     *                     <pre>
     *                                             HBaseUtil.deleteColumn("TEST", "row1", "info", "name");
     *                                         </pre>
     */
    public static void deleteColumn(String namespace, String tableName, String rowKey, String columnFamily, String columnName) throws IOException {
        validateInputs(tableName, rowKey, columnFamily, columnName);
        try (Connection conn = getConnection(); Table table = conn.getTable(TableName.valueOf(namespace, tableName))) {
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            delete.addColumns(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName));
            table.delete(delete);
        }
    }

    /**
     * 主方法，测试 HBaseUtil 的所有功能。
     *
     * @param args 命令行参数（未使用）
     */
    public static void main(String[] args) throws IOException {

        try {
            // 测试 DDL 操作
            System.out.println("=== 开始 DDL 测试 ===");
            // 检查表是否存在
            boolean exists = isTableExists("LOCOMOTIVE_ADS", "TEST");
            System.out.println("exists = " + exists);
            // 删除表
            deleteTable("LOCOMOTIVE_ADS", "TEST");
            // 创建表,并指定列族
            createTable("LOCOMOTIVE_ADS", "TEST", new String[]{"info", "address", "job"});
            putCell("LOCOMOTIVE_ADS", "TEST", "HXD1D0001", "info", "name", "tiancy");
            putCell("LOCOMOTIVE_ADS", "TEST", "HXD1D0002", "info", "name", "帅哥");
            putCell("LOCOMOTIVE_ADS", "TEST", "HXD1D0001", "address", "city", "北京");
            putCell("LOCOMOTIVE_ADS", "TEST", "HXD1D0001", "job", "zw", "大数据");
            List<JSONObject> cells = getCells("LOCOMOTIVE_ADS", "TEST", "HXD1D0001", "info", "name");
            cells.forEach(System.out::println);
            List<JSONObject> jsonObjects = scanRows("LOCOMOTIVE_ADS", "TEST", "HXD1D0001", "HXD1D0002");
            jsonObjects.forEach(System.out::println);
            String family = "DATA";
            String[] columnsBd = {"JD", "WD"}; // 列名
            List<JSONObject> jsonObjects1 = queryHBaseTable("LOCOMOTIVE_ODS:CMD_ITF_BD_HEADER", "23900012024-10-29 09:57:00", "23900012025-04-23 15:58:00", family, columnsBd);
            System.out.println("jsonObjects1 = " + jsonObjects1);
        } finally {
            closeConnection();
        }
    }
}